package com.semidata.rtd.core.executor;

import java.util.concurrent.Callable;
import java.util.concurrent.Future;
import java.util.concurrent.FutureTask;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RunnableFuture;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import com.semidata.rtd.core.job.Task;

public class ThreadPoolTaskExecutor extends ThreadPoolExecutor implements TaskExecutor {

    private static final long KEEP_ALIVE_TIME = 10;
    private static final TimeUnit TIME_UNIT = TimeUnit.SECONDS;
    private static final int CORE_POOL_SIZE = 16;
    private static final int MAX_POOL_SIZE = Integer.MAX_VALUE;
    

    public ThreadPoolTaskExecutor() {
        super(CORE_POOL_SIZE, MAX_POOL_SIZE, KEEP_ALIVE_TIME, TIME_UNIT, new LinkedBlockingQueue<Runnable>());
    }

    private static class SubmittedTask<T> implements Callable<T> {
        
        private Task<T> task;
        private TaskCompleteHandler<T> handler;

        public SubmittedTask(Task<T> task, final TaskCompleteHandler<T> handler) {
            this.task = task;
            this.handler = handler;
        }
        
        @Override
        public T call() throws Exception {
            return task.run();
        }
       
        public TaskCompleteHandler<T> getTaskCompleteHandler() {
            return handler;
        }

        public Task<T> getTask() {
            return task;
        }
    }
    
    private static class SubmittedTaskWrapper<T> extends FutureTask<T> {

        private SubmittedTask<T> submittedTask;

        public SubmittedTaskWrapper(SubmittedTask<T> submittedTask) {
            super(submittedTask);
            this.submittedTask = submittedTask;
        }
        
        public TaskCompleteHandler<T> getTaskCompleteHandler() {
            return submittedTask.getTaskCompleteHandler();
        }
        
        public Task<T> getTask() {
            return submittedTask.getTask();
        }
        
    }
    
    @Override
    public <T> Future<T> submit(Task<T> task, TaskCompleteHandler<T> handler) {
        return submit(new SubmittedTask<T>(task, handler));
    }
    
    @SuppressWarnings({ "rawtypes", "unchecked" })
    @Override
    protected void afterExecute(Runnable runnable, Throwable throwable) {

        if (runnable instanceof SubmittedTaskWrapper) {
            SubmittedTaskWrapper future = (SubmittedTaskWrapper) runnable;
            
            
            Throwable cause = throwable;
            
            Task task = future.getTask();
            TaskCompleteHandler handler = future.getTaskCompleteHandler();
            
            try {
                handler.handle(task, future.get(), cause);
            } catch (Exception e) {
                handler.handle(task, null, e);
            }
        }
    }

    @Override
    protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
        if (callable instanceof SubmittedTask)
            return new SubmittedTaskWrapper<T>((SubmittedTask<T>) callable);
        return super.newTaskFor(callable);
    }

    @Override
    public void start() {
    }

    @Override
    public void stop() {
        shutdown();
    }

    
}
